package com.hliushi.spark.rdd

import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/5/15 9:25
 */
class ActionOp {

  /**
   * Action算子动作会触发一次Job的运行
   * 默认情况下, 每一个Action运行的时候, 其所关联的所有Transformation RDD 都要重新计算
   * 但是也可以使用presist方法, 将RDD持久化到磁盘或者内存中, 这个时候下次可以更快的访问
   */
  val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("transformation_op")
  val sc = new SparkContext(conf)


  /**
   * 需求: 最终生成("结果", price)
   *
   * 函数原型
   * def reduce(f: (T, T) => T): T
   * 注意点:
   * 1.函数传入的curr参数, 并不是value, 而是一整条数据
   * 2.reduce整体上的结果, 只有一个
   *
   */
  @Test
  def reduce(): Unit = {
    val rdd = sc.parallelize(Seq(("手机", 10.0), ("手机", 15.0), ("电脑", 20.0)))

    val result = rdd.
      reduce((total: (String, Double), curr: (String, Double)) => ("总价", total._2 + curr._2))

    println(result)
  }


  /**
   * 函数原型
   * def foreach(f: T => Unit): Unit
   */
  @Test
  def foreach(): Unit = {
    val rdd = sc.parallelize(Seq(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
    rdd.foreach(x => print(x + " "))

    /**
     * 结果
     * 2 1 3 9 7 4 6 5 8 10
     * 从这个打印顺序是无序的, 可以看出这个foreach是一个 异步过程[aysn]
     */
  }


  /**
   * count和countByKey的结果相距很远很远, 每次调用Action都会生成一个job, job会获取运行结果
   * 所以在两个job中间有大量的Log打出, 其实就是在启动job
   * -
   * countByKey的运算结果是Map(key, value -> key的count)
   * -
   * 数据倾斜, 如果要解决数据倾斜的问题, 是不是要先知道谁倾斜,
   * 通过countByKey是不是可以查看key对应的数据总数, 从而解决倾斜问题
   * -
   * 函数原型
   * def count(): Long
   * #
   * def countByKey(): Map[K, Long]
   */
  @Test
  def count(): Unit = {
    val rdd = sc.parallelize(Seq(("a", 1), ("b", 2), ("d", 3), ("e", 5), ("f", 6), ("a", 6)))

    // rdd.count() = 6
    println(s"rdd.count() = ${rdd.count()}")
    // rdd.countByKey() = Map(e -> 1, f -> 1, a -> 2, b -> 1, d -> 1)
    println(s"rdd.countByKey() = ${rdd.countByKey()}")
  }


  /**
   * take和takeSample都是获取数据, 一个是直接获取,
   * .      一个是采样获取[类似sample, 区别在这是一个Action, 直接返回数据]
   * first: 一般情况下, action会从所有分区获取数据, 相对来说速度就比较慢
   * .      但first只是获取第一个元素, 所以first只会处理第一个分区, 速度很快, 无须处理所有数据
   *
   * 函数原型
   * def first(): T
   * #
   * def take(num: Int): Array[T]
   * #
   * def takeSample(
   * .    withReplacement: Boolean,
   * .    num: Int,
   * .    seed: Long = Utils.random.nextLong): Array[T]
   */
  @Test
  def take(): Unit = {
    val rdd = sc.parallelize(Seq(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))

    // rdd.first() = 1
    println(s"rdd.first() = ${rdd.first()}")
    // 1 2 3
    println(s"rdd.take(3)")
    rdd.take(3).foreach(println(_))

    // 2 10 8 8 1 4 7
    rdd.takeSample(withReplacement = true, num = 7).foreach(println(_))
  }


  /**
   * 指定初始化和计算函数, 折叠集合整个数据集
   * 函数原型
   * def fold(zeroValue: T)(op: (T, T) => T): T
   */
  @Test
  def fold(): Unit = {
    val array = Seq(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val rdd = sc.parallelize(array)

    println(s"rdd.getNumPartitions = ${rdd.getNumPartitions}")

    val foldResult = rdd.fold(100)((total: Int, curr: Int) => total + curr)

    /**
     * foldResult = 755
     * 解释:
     * SparkConf().setMaster("local[6]")  可以得出, rdd在没有手动添加分区时, 会按照这个参数默认分配6个分区
     * 755 = 6 * 100 + 100 + 55
     *
     */
    println(s"foldResult = ${foldResult}")
  }


  /**
   *
   */
  @Test
  def numeric(): Unit = {
    val array = Seq(1, 2, 3, 4, 10, 20, 30, 50, 100)
    val rdd = sc.parallelize(array)

    println(s"rdd.max() = ${rdd.max()}")
    println(s"rdd.min() = ${rdd.min()}")
    // mean 是平均数  rdd.mean() = 24.444444444444443
    println(s"rdd.mean() = ${rdd.mean()}")
    println(s"rdd.sum() = ${rdd.sum()}")

    // variance 是方差     rdd.variance() = 950.2469135802468
    println(s"rdd.variance() = ${rdd.variance()}")
    // stdev 是标准差       rdd.stdev() = 30.82607522180284
    println(s"rdd.stdev() = ${rdd.stdev()}")
  }
}